package com.atguigu.edu.realtime.app.dim.func;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.edu.realtime.bean.TableProcess;
import com.atguigu.edu.realtime.constant.EduConfig;
import com.atguigu.edu.realtime.util.PhoenixUtil;
import org.apache.flink.api.common.state.BroadcastState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.state.ReadOnlyBroadcastState;
import org.apache.flink.streaming.api.functions.co.BroadcastProcessFunction;
import org.apache.flink.util.Collector;

import java.util.HashSet;
import java.util.Map;
import java.util.Set;

public class TableProcessFunction extends BroadcastProcessFunction<JSONObject,String,JSONObject> {

    //声明状态//增加构造函数->为方便写入状态
    MapStateDescriptor<String, TableProcess> tableProcessMapStateDescriptor;
    public TableProcessFunction(MapStateDescriptor<String, TableProcess> tableProcessMapStateDescriptor){
        this.tableProcessMapStateDescriptor = tableProcessMapStateDescriptor;
    }
    /*    {
            "database": "gmall",
                "xid": 35197,
                "data": {
            "birthday": "2001-03-10",
                    "gender": "M",
                    "create_time": "2020-06-10 10:56:50",
                    "login_name": "ufl58mym",
                    "nick_name": "阿家",
                    "name": "汤亨",
                    "user_level": "1",
                    "phone_num": "13585418286",
                    "id": 1,
                    "email": "ufl58mym@msn.com",
                    "operate_time": "2020-06-15 13:48:50"
        },
            "old": {
            "nick_name": "阿家11"
        },
            "commit": true,
                "type": "update",
                "table": "user_info",
                "ts": 1665656486
        }*/
    //主流数据到达时要做的处理 ：根据广播流的配置表状态信息进行筛选
    @Override
    public void processElement(JSONObject eduDataJson, ReadOnlyContext ctx, Collector<JSONObject> out) throws Exception {
        //System.out.println("主流数据" + gmallDataJson);
        //得到状态
        ReadOnlyBroadcastState<String, TableProcess> broadcastState = ctx.getBroadcastState(tableProcessMapStateDescriptor);
        //1.判断主流数据的表是否在状态中，在-->保留，不在-->丢弃
        //提取主流数据的表跟TableProcess做比较
        String tableName = eduDataJson.getString("table");
        TableProcess tableProcess = broadcastState.get(tableName);
        String optType = eduDataJson.getString("type");
        if (tableProcess != null&&!optType.equals("bootstrap-start")&&!optType.equals("bootstrap-complete")){
            //2.如果保留，剪裁字段：根据配置表中规定的字段保留，把不需要的字段删除
            String sinkColumns = tableProcess.getSinkColumns();//首先将表格的字段先取出来
            String[] colArr = sinkColumns.split(","); //切分字段以便于进行比对
            Set colset = new HashSet<String>();//将colArr转化为set
            for (String col : colArr) {
                colset.add(col);  //完成集合
            }
            //判断数据的每个字段是不是在集合里
            JSONObject dataJsonObj = eduDataJson.getJSONObject("data");
            Set<Map.Entry<String, Object>> dataKeyValue = dataJsonObj.entrySet();//每个字段怎么取？
            //对字段是否删除进行判断
            dataKeyValue.removeIf(entry ->!colset.contains(entry.getKey()));//如果包含就不删除->!
            //3.补充要真正写入到phoenix表的表名
            eduDataJson.put("sink_table",tableProcess.getSinkTable());
            //4.输出到out中:打印
            out.collect(eduDataJson);
        }
    }

    //广播流数据到达时，要做的处理 ：更新广播状态
    @Override
    public void processBroadcastElement(String tableProcessJson, Context ctx, Collector<JSONObject> out) throws Exception {
        //广播流数据{"before":null,
        // "after":{"source_table":"base_region","sink_table":"dim_base_region","sink_columns":"id,region_name","sink_pk":"id","sink_extend":null},
        // "source":{"version":"1.5.4.Final","connector":"mysql","name":"mysql_binlog_source","ts_ms":1665483940304,"snapshot":"false","db":"gmall_realtime_config","sequence":null,"table":"table_process","server_id":0,"gtid":null,"file":"","pos":0,"row":0,"thread":null,"query":null},"op":"r","ts_ms":1665483940304,"transaction":null}
        //System.out.println("广播流数据" + tableProcessJson);

        //1. 将json转换成jsonObj
        JSONObject tableProcessJsonObj = JSON.parseObject(tableProcessJson);
        //2. 提取操作符
        String operator = tableProcessJsonObj.getString("op");

        //获得广播流的状态
        BroadcastState<String, TableProcess> broadcastState = ctx.getBroadcastState(tableProcessMapStateDescriptor);

        //3. 区分广播流的操作类型
        if (!operator.equals("d")){ //不是删除的时候
            //3.1  如果是新增+修改+全量查询，直接维护进状态
            //3.1.1 先把after中的数据提取出来，转成JavaBean TableProcess
            TableProcess tableProcess = tableProcessJsonObj.getObject("after", TableProcess.class);
            //3.1.2  检查phoenix置否有该维表，如果没有生成建表语句，执行建表
            checkTable(tableProcess);
            //3.1.3  写入状态  getSourceTable:为了让数据未来可以找到它的相关的配置信息
            broadcastState.put(tableProcess.getSourceTable(),tableProcess);
        }else{ //3.2  如果是删除，要从状态中移除
            //首先得到要删除的信息：要删除哪张表
            TableProcess tableProcess = tableProcessJsonObj.getObject("before", TableProcess.class);
            //根据表名去掉状态
            broadcastState.remove(tableProcess.getSourceTable());
        }
    }

    private void checkTable(TableProcess tableProcess){
        // create table if not exists  source_table
        // (  col1  varchar  [primary key],  //根据sink_columns 进行拼接
        //    col2  varchar ...
        // ) [SALT_BUCKETS = 3]  //延迟分桶的操作

        //尝试进行拼接表
        StringBuilder createTableSQL =
                new StringBuilder("create table if not exists " + EduConfig.PHOENIX_SCHEMA + "." + tableProcess.getSinkTable()+"(");
        //数据原本是json的字符串，需要拆开，此处是拆开 列
        String[] sinkCols = tableProcess.getSinkColumns().split(",");
        //迭代  itar
        for (int i = 0; i < sinkCols.length; i++) {
            String sinkCol = sinkCols[i];
            if (i>0){
                createTableSQL.append(", ");
            }
            createTableSQL.append(sinkCol+" varchar ");
            //加不加[primary key]取决于列名sinkCol是否与sink_pk一样
            if (sinkCol.equals(tableProcess.getSinkPk())){
                createTableSQL.append(" primary key ");
            }
        }

        createTableSQL.append(")");
        //判断要不要拼接 sink_extend
        if (tableProcess.getSinkExtend()!=null&&tableProcess.getSinkExtend().trim().length()>0){
            createTableSQL.append(tableProcess.getSinkExtend());
        }
        System.out.println(createTableSQL.toString());
        PhoenixUtil.executeSql(createTableSQL.toString());
    }
}

